package com.rr.learningdemo.rabbitmq.listener;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.rr.learningdemo.rabbitmq.thread.RedisListenerThread;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SetOperations;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;

/**
 * @author zhuangcanrun
 * @Date 2021/06/21/16:51
 * @description
 **/

@Component
public class RabbitMQListener {

    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private RedisListenerThread redisListenerThread;

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = "queueForDistinct", durable = "true"),
                    exchange = @Exchange(value = "exchangeForDistinct", type = ExchangeTypes.DIRECT, durable = "true"),
                    key = "queueForDistinct"
            )
    )
    private void onMessage(Message message, Channel channel) throws IOException {
        String newMsg = new String(message.getBody(), StandardCharsets.UTF_8);
        List<String> strings = Arrays.asList(newMsg.split(","));
        SetOperations setOperations = redisTemplate.opsForSet();
        for (String string : strings) {
            if (!setOperations.isMember("set", string)) {
                System.out.println("插入:" + string);
                setOperations.add("set", string);
            }else {
                System.out.println(string + "已存在队列");
            }
        }
        redisListenerThread.resumeThread();
    }
}
